package org.geekbang.projects.messaging;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.logging.Level;
import java.util.logging.Logger;

/**
* 订阅者包委托处理器
* @author : KangNing Hu
*/
public  class SubscriptionDelegate<T> implements Subscription , Subscriber<T> {


	private static final Logger logger = Logger.getLogger(SubscriptionDelegate.class.getName());
	/**
	 * 上下文
	 */
	private StreamContext<T> streamContext;

	/**
	 * 被委托的订阅者
	 */
	private final Subscriber<? super T> subscriber;

	/**
	 * 请求计数器
	 */
	private long requestCount = 1;

	/**
	 * 最大请求量
	 */
	private volatile long maxRequestCount = -1;

	/**
	 * 订阅者取消状态
	 * 如果为true则取消订阅
	 * 如果为false则为正常订阅状态
	 */
	private volatile boolean cancelState = false;


	/**
	 * 是否被订阅被注销
	 * 如果为true则已被注销
	 * 如果为false则未被注销
	 */
	private volatile boolean unsubscribe = false;


	public SubscriptionDelegate(Subscriber<? super T> subscriber , StreamContext<T> streamContext ){
		this.subscriber = subscriber;
		this.streamContext = streamContext;
	}


	/**
	 * 获取原始的订阅者
	 * @return
	 */
	public Subscriber<? super T> getRawSubscriber(){
		return this.subscriber;
	}

	@Override
	public void onSubscribe(Subscription s) {
		subscriber.onSubscribe(s);
	}

	@Override
	public void onNext(T o) {
		if (this.unsubscribe){
			logger.log(Level.WARNING , String.format("%s订阅已是注销状态" , this.subscriber.getClass().getName()));
			return;
		}
		try {
			synchronized (this.subscriber) {
				//校验请求
				checkRequest();
				//通知目标订阅者
				subscriber.onNext(o);
				//更新请求次数
				requestCount++;
			}
		}catch (Throwable throwable){
			onError(throwable);
		}finally {
			//如过当前订阅已完整则回调
			if (this.cancelState && !this.unsubscribe){
				onComplete();
				synchronized (this) {
					streamContext.unsubscribe(this);
					unsubscribe = true;
				}
			}
		}

	}


	@Override
	public void onError(Throwable t) {
		subscriber.onError(t);
	}

	@Override
	public void onComplete() {
		subscriber.onComplete();
	}

	@Override
	public void request(long n) {
		this.maxRequestCount = n;
	}

	@Override
	public void cancel() {
		this.cancelState = true;
	}

	/**
	 * 是否取消
	 * @return
	 */
	public boolean isCancelState(){
		return this.cancelState;
	}

	/**
	 * 校验请求
	 */
	private void checkRequest() {
		//如果为-1 则不限制请求量
		if (this.maxRequestCount == -1){
			return;
		}
		//校验请求量是否益出
		if (this.requestCount > maxRequestCount){
			this.cancelState = true;
			throw new IllegalStateException(String.format("当前订阅大于请求量 %s" , this.maxRequestCount));
		}

	}
}
